package com.xinqing.bigdata.test.datastream

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._


object WC {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val params: ParameterTool = ParameterTool.fromArgs(args)
    val input: String = params.get("input")
    val output: String = params.get("output")
    //source算子
    val inputStream = env.readTextFile(input)

    inputStream.map(data => {
      val dataRow = data.split(",")
      SensorReading(dataRow(0).trim, dataRow(1).trim.toLong, dataRow(2).trim.toDouble)
    }).keyBy(0).sum(1).writeAsText(output)

    env.execute("wc job")
  }
}
